#pragma once
#include "rpc_router.hpp"
#include "rpc_register.hpp"
#include "../common/dispatcher.hpp"
#include "../client/rpc_client.hpp"
#include <memory>
#include <unordered_set>

namespace RPC
{
    namespace Server
    {
        class TopicManager
        {
        public:
            using Ptr = std::shared_ptr<TopicManager>;

            // 进行主题操作
            void onTopicOperate(const BaseConnection::Ptr &conn, const TopicRequest::Ptr &msg)
            {

                bool ret = true;
                switch (msg->opType())
                {
                case TopicOpType::TOPIC_CREAT:
                    topicCreate(msg);
                    break; // 创建主题
                case TopicOpType::TOPIC_DELETE:
                    topicOffline(msg);
                    break; // 主题删除
                case TopicOpType::TOPIC_SUB:
                    ret = topicSubscribe(conn, msg);
                    break; // 主题订阅
                case TopicOpType::TOPIC_UNSUB:
                    topicCancel(conn, msg);
                    break; // 主题取消订阅
                case TopicOpType::TOPIC_PUBLISH:
                    ret = topicPublish(conn, msg);
                    break; // 主题发布
                default:
                    return errorResponse(conn, msg, RCode::RCODE_INVALID_TOPIC_OPE);
                }
                if (ret)
                    return normalResponse(conn, msg);
                else
                    return errorResponse(conn, msg, RCode::RCODE_NOT_TOPIC);
            }

            void errorResponse(const BaseConnection::Ptr &conn, const TopicRequest::Ptr &msg, const RCode &code)
            {
                TopicResponse::Ptr topicrsp = MessageFactory::create<TopicResponse>();
                topicrsp->setId(msg->id());
                topicrsp->setRCode(code);
                topicrsp->setType(MType::RES_TOPIC);
                conn->send(topicrsp);
            }

            void normalResponse(const BaseConnection::Ptr &conn, const TopicRequest::Ptr &msg)
            {
                TopicResponse::Ptr topicrsp = MessageFactory::create<TopicResponse>();
                topicrsp->setId(msg->id());
                topicrsp->setRCode(RCode::RCODE_OK);
                topicrsp->setType(MType::RES_TOPIC);
                conn->send(topicrsp);
            }
            // 连接断开时进行操作
            void onShutdown(const BaseConnection::Ptr &conn)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                {
                    // 1、找到连接对应的subscribe
                    auto conn_it = _conn_subs.find(conn);
                    if (conn_it == _conn_subs.end())
                        return;
                    Subscribe::Ptr subscribe = conn_it->second;
                    // 2、遍历subscribe的topic，去每个topic里面删除该连接
                    for (auto topic : subscribe->topics)
                    {
                        // //找不到对应的主题，直接返回，否则向下会创建主题，插入pair(topic, nullptr)
                        // if(_topic_conns.count(topic) == 0) continue;
                        // _topic_conns[topic]->remove(subscribe);

                        auto topic_it = _topic_conns.find(topic);
                        if (topic_it == _topic_conns.end())
                            continue;
                        topic_it->second->remove(subscribe);
                    }
                    // 3、删除连接
                    _conn_subs.erase(conn);
                }
            }

        private:
            //  后续如要在函数里面发送消息，就加上conn
            void topicCreate(const TopicRequest::Ptr &msg)
            {
                // 创建主题
                Topic::Ptr topic;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _topic_conns.find(msg->topicKey());
                    if (it != _topic_conns.end())
                    {
                        ERROR_LOG("该主题已经创建！");
                        return;
                    }
                    topic = std::make_shared<Topic>(msg->topicKey());
                    _topic_conns.insert({msg->topicKey(), topic});
                }
            }

            void topicOffline(const TopicRequest::Ptr &msg)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _topic_conns.find(msg->topicKey());
                if (it == _topic_conns.end())
                {
                    ERROR_LOG("该主题已经不存在！");
                    return;
                }
                // 拿到将要下线的主题
                Topic::Ptr topicmember = it->second;
                // 找到下线主题里面的所有连接
                for (auto &con : topicmember->conns)
                {
                    if (!_conn_subs.count(con->conn))
                        continue;
                    // 找到连接，删除对应主题
                    _conn_subs[con->conn]->remove(msg->topicKey());
                }
                // 删除主题
                _topic_conns.erase(msg->topicKey());
                return;
            }

            bool topicSubscribe(const BaseConnection::Ptr &conn, const TopicRequest::Ptr &msg)
            {
                Subscribe::Ptr subscribe;
                Topic::Ptr topic;
                std::unique_lock<std::mutex> lock(_mutex);
                {
                    // 1、找到对应的主题，没有报错返回
                    auto topic_it = _topic_conns.find(msg->topicKey());
                    if (topic_it == _topic_conns.end())
                    {
                        ERROR_LOG("没有对应的主题！");
                        return false;
                    }

                    topic = topic_it->second;
                    // 2、找到对应的连接，没有创建
                    auto conn_it = _conn_subs.find(conn);
                    if (conn_it == _conn_subs.end())
                    {
                        subscribe = std::make_shared<Subscribe>(conn);
                        _conn_subs.insert({conn, subscribe});
                    }
                    else 
                        subscribe = conn_it->second;
                }
                // 3、在连接者中添加订阅的主题
                subscribe->append(msg->topicKey());
                // 4、在主题中添加连接者
                topic->append(subscribe);
                return true;
            }

            void topicCancel(const BaseConnection::Ptr &conn, const TopicRequest::Ptr &msg)
            {
                Subscribe::Ptr subscribe;
                Topic::Ptr topic;
                std::unique_lock<std::mutex> lock(_mutex);
                {
                    // 1、找到对应的主题，没有报错返回
                    auto topic_it = _topic_conns.find(msg->topicKey());
                    if (topic_it != _topic_conns.end())
                    {
                        topic = topic_it->second;
                    }
                    // 2、找到对应的连接，没有创建
                    auto conn_it = _conn_subs.find(conn);
                    if (conn_it != _conn_subs.end())
                    {
                        subscribe = conn_it->second;
                    }
                }
                if (topic)
                    topic->remove(subscribe);
                if (subscribe)
                    subscribe->remove(msg->topicKey());
            }

            bool topicPublish(const BaseConnection::Ptr &conn, const TopicRequest::Ptr &msg)
            {
                Topic::Ptr topic;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topic_conns.find(msg->topicKey());
                    if (topic_it == _topic_conns.end())
                        return false;
                    topic = topic_it->second;
                }
                topic->publish(msg);
                return true;
            }

            struct Subscribe // 提供下线基础，上线的时候需要维护
            {
                using Ptr = std::shared_ptr<Subscribe>;

                Subscribe(const BaseConnection::Ptr &c) : conn(c)
                {
                }
                void append(const std::string &topic)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    topics.insert(topic);
                }

                void remove(const std::string &topic)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    topics.erase(topic);
                }

                std::mutex mutex;
                BaseConnection::Ptr conn;
                std::unordered_set<std::string> topics;
            };
            struct Topic // 提供上线下线基础
            {
                using Ptr = std::shared_ptr<Topic>;

                Topic(const std::string &topic) : topic_name(topic)
                {
                }

                void append(const Subscribe::Ptr &conn)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    conns.insert(conn);
                }

                void remove(const Subscribe::Ptr &conn)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    conns.erase(conn);
                }

                void publish(const BaseMessage::Ptr &msg)
                {
                    for (const auto &c : conns)
                    {
                        c->conn->send(msg);
                    }
                }

                std::mutex mutex;
                std::string topic_name;
                std::unordered_set<Subscribe::Ptr> conns;
            };

        private:
            std::mutex _mutex;
            std::unordered_map<std::string, Topic::Ptr> _topic_conns;
            std::unordered_map<BaseConnection::Ptr, Subscribe::Ptr> _conn_subs;
        };
    }
}